package com.bclz.parallelism

import java.util.concurrent.{Callable, ExecutorService, Executors, Future, TimeUnit}



case class UnitFuture[A](value:A) extends Future[A] {
  override def cancel(mayInterruptIfRunning: Boolean): Boolean = false

  override def isCancelled: Boolean = false

  override def isDone: Boolean = true

  override def get(): A = value

  override def get(timeout: Long, unit: TimeUnit): A = value
}

/**
  * map2 Future ,支持超时设置
  */
case class map2Future[A,B,C](a:Future[A],b:Future[B],f:(A,B)=>C) extends Future[C]{
  //定义一个cache缓存计算值，避免多线程执行造成结果不一致
  @volatile var cache:Option[C]=None

  override def cancel(mayInterruptIfRunning: Boolean): Boolean = a.cancel(mayInterruptIfRunning) || b.cancel(mayInterruptIfRunning)

  override def isCancelled: Boolean = a.isCancelled || b.isCancelled

  override def isDone: Boolean = cache.isDefined

  override def get(): C = compute(Long.MaxValue)

  override def get(timeout: Long, unit: TimeUnit): C =
    compute(TimeUnit.NANOSECONDS.convert(timeout,unit))

  def compute(timeoutOfNanos:Long):C= cache match {
    case Some(x) => x
    case None => {

      val startTime=System.nanoTime
      val av = a.get(timeoutOfNanos,TimeUnit.NANOSECONDS)
      val stopTime=System.nanoTime
      val duration=stopTime-startTime
      val bv=b.get(timeoutOfNanos-duration,TimeUnit.NANOSECONDS)
      val cv=f(av,bv)
      cache=Some(cv)
      cv

    }

  }

}

object Par  {
  //抽象一个并行函数模型
  type Par[A]= ExecutorService=>Future[A]
//  def unit[A](a: =>A):Par[A]= ???

  //unit参数是一种严格求值，直接获取结果.计算单元
  def unit[A](a:A):Par[A]= _=>UnitFuture(a)

  //fork提供并行计算的能力函数
  //该实现存在问题，get会阻塞线程
  def fork[A](p: =>Par[A]):Par[A]=e=>e.submit(new Callable[A] {
    override def call(): A = p(e).get()
  })

  //lazyUnit参数惰性值，可并行求值
  def lazyUnit[A](a: =>A):Par[A]=fork(unit(a))

  //合并两个par结果 的能力
  def map2[A,B,C](pa:Par[A],pb:Par[B])(f:(A,B)=>C):Par[C] =
    e=>e.submit(new Callable[C] {
      override def call(): C = f(pa(e).get,pb(e).get)
    })

  //真正执行，获取结果
  def run[A](e:ExecutorService)(pa:Par[A]):Future[A]=pa(e)


  def asyncF[A,B](f:A=>B):A=>Par[B]=
      a => lazyUnit(f(a))

  //隐式转换，可以是type Par像普通对象一样使用前缀表达式
  implicit def toParOps[A](p: Par[A]): ParOps[A] = new ParOps(p)

  class ParOps[A](p: Par[A]) {
    def map2[B,C](pb:Par[B])(f:(A,B)=>C):Par[C] =
      e=>e.submit(new Callable[C] {
        override def call(): C = f(p(e).get,pb(e).get)
      })

    def map[B](f:A=>B):Par[B]=
      p.map2(unit())((a,_)=>f(a))

  }


  def sequenceViaFoldRight[A](pa:List[Par[A]]):Par[List[A]]=
    pa.foldRight[Par[List[A]]](unit(List()))(_.map2(_)(_::_))

  //将递归计算放在新的逻辑线程，使有效的尾递归
  //
  def sequenceViaFork[A](pa:List[Par[A]]):Par[List[A]]=pa match {
    case Nil => unit(Nil)
    case h::t=>h.map2(fork(sequenceViaFork(t)))(_::_)
  }

  //平分一半已获得更好的性能,然后派生N个并行计算，1/2->1/4->...->1/n,线程池中异步计算
  def sequenceBalance[A](pa:IndexedSeq[Par[A]]):Par[IndexedSeq[A]]=fork {

    if(pa.isEmpty) unit(Vector())
    else if(pa.length==1) pa.head.map(Vector(_))
    else {
      val (p1,p2) = pa.splitAt(pa.length/2)
      fork(sequenceBalance(p1).map2(sequenceBalance(p2))(_ ++ _))
    }

  }

  /**
    * 并行map
    */
  def parMap[A,B](as:List[A])(f:A=>B):Par[List[B]]= fork{
    val listParB = as.map(asyncF(f))
    sequenceViaFork(listParB)
  }

  def parFilter[A](as:List[A])(f:A=>Boolean):Par[List[A]]= {
    val list = as.map(asyncF(a => if (f(a)) List(a) else List()))
    sequenceViaFork(list).map(_.flatten)  //flatten合并集合
  }


  def parReduceLeft[A](ints:IndexedSeq[A])(f:(A,A)=>A):Par[A]={
    if(ints.isEmpty)
      throw new UnsupportedOperationException("empty.parReduceLeft")
    else if(ints.length==1) unit(ints.head)
    else {
      val (l1,l2) = ints.splitAt(ints.length/2)
      fork(parReduceLeft(l1)(f)).map2(fork(parReduceLeft(l2)(f)))(f(_,_))
    }

  }



  def main(args: Array[String]): Unit = {
    val pool=Executors.newCachedThreadPool
    //如果只有一个线程，会无期限等待，fork实现会提交Par在子线程中，
    // 如果是parallelism par就会提交一个任务等待空余的线程,
    // 而fork提交的线程Callable也会等待par的Callable执行结果(future get会一直等待阻塞),死锁就发生了
    //这个简单的fork实现有一定的缺陷
//val pool=Executors.newSingleThreadExecutor()
    //定义一个求和函数 ，采用并行计算
    def sum(ints:IndexedSeq[Int]):Par[Int]={
      if(ints.size<=1) unit(ints.headOption getOrElse(0) )
      else{
        val (one,two) = ints.splitAt(ints.length/2)
//        map2(fork(sum(one)),fork(sum(two)))(_+_)
//        map2(sum(one),sum(two))(_+_)
        sum(one).map2(sum(two))(_+_)    //隐式转换toParOps
      }
    }
//    val evenFuture = run(pool)(parFilter((1 to 10000).toList)(i=>(i&1)==0))
//    val sumFuture=run(pool)(sum(1 to 1000000))
    val sumFuture2=run(pool)(parReduceLeft(1 to 1000000)(_+_))
//val maxFuture=run(pool)(parReduceLeft(1 to 100)(_ max _))
//    println(maxFuture.get())
//      println((1 to 1000000).sum)
//    println(sumFuture.get())
    println(sumFuture2.get())
    pool.shutdown()
  }
}
